package com.ayu.shortlink.distribute.mq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson2.JSON;
import com.ayu.shortlink.distribute.common.constant.RedisCacheConstant;
import com.ayu.shortlink.distribute.common.constant.RocketMQConstant;
import com.ayu.shortlink.distribute.common.convention.exception.ClientException;
import com.ayu.shortlink.distribute.common.enums.CouponTaskStatusEnum;
import com.ayu.shortlink.distribute.common.enums.UserCouponStatusEnum;
import com.ayu.shortlink.distribute.dao.entity.CouponTaskDO;
import com.ayu.shortlink.distribute.dao.entity.CouponTemplateDO;
import com.ayu.shortlink.distribute.dao.entity.UserCouponDO;
import com.ayu.shortlink.distribute.dao.mapper.CouponTaskMapper;
import com.ayu.shortlink.distribute.dao.mapper.CouponTemplateMapper;
import com.ayu.shortlink.distribute.dao.mapper.UserCouponMapper;
import com.ayu.shortlink.distribute.mq.basics.base.MessageWrapper;
import com.ayu.shortlink.distribute.mq.basics.event.CouponTemplateExecuteEvent;
import com.ayu.shortlink.distribute.mq.basics.event.UserCouponOnTimeCloseEvent;
import com.ayu.shortlink.distribute.mq.producer.UserCouponOnTimeCloseProducer;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.toolkit.SqlHelper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.executor.BatchExecutorException;
import org.apache.ibatis.executor.BatchResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

/**
 * 执行刷盘的消费者 very Important
 */
@Component
@RequiredArgsConstructor
@Slf4j
@RocketMQMessageListener(
        topic = RocketMQConstant.TOPIC_ACTUALLY_EXECUTOR_TASK,
        consumerGroup = RocketMQConstant.GROUP_ACTUALLY_EXECUTOR_TASK
)
public class CouponTaskActuallyDistributeConsumer implements RocketMQListener<MessageWrapper<CouponTemplateExecuteEvent>> {
    private final StringRedisTemplate stringRedisTemplate;
    private final CouponTemplateMapper couponTemplateMapper;
    private final UserCouponMapper userCouponMapper;
    private final CouponTaskMapper couponTaskMapper;
    private final UserCouponOnTimeCloseProducer userCouponOnTimeCloseProducer;
    // 期望执行批量操作的数量
    private static final Integer USER_SIZE_EXPECT = 5000;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void onMessage(MessageWrapper<CouponTemplateExecuteEvent> messageWrapper) {

        log.info("正在执行优惠券扣减和用户获取优惠券:{}", JSON.toJSONString(messageWrapper));
        CouponTemplateExecuteEvent message = messageWrapper.getMessage();
        String UserBatchSizeKey = String.format(RedisCacheConstant.COUPON_DISTRIBUTE_BATCH_USER_SAVE_KEY, message.getCouponTaskId());
        //获取用户数量
        Long userBatchSize = stringRedisTemplate.opsForSet().size(UserBatchSizeKey);

        if(!message.getDistributionEndFlag() && userBatchSize!= null && userBatchSize >= USER_SIZE_EXPECT){
            // 执行优惠券扣减和用户获取优惠券的功能
            deductStockAndUserGetCoupon(message, userBatchSize);
        }

        // 进行兜底，防止优惠券不满足5000
        if(message.getDistributionEndFlag()){
            Long couponTemplateId = Long.valueOf(message.getCouponTemplateId());
            LambdaQueryWrapper<CouponTemplateDO> queryWrapper = Wrappers.lambdaQuery(CouponTemplateDO.class)
                    .eq(CouponTemplateDO::getShopNumber, message.getShopNumber())
                    .eq(CouponTemplateDO::getId, couponTemplateId);
            CouponTemplateDO couponTemplateDO = couponTemplateMapper.selectOne(queryWrapper);
            deductStockAndUserGetCoupon(message, Long.valueOf(couponTemplateDO.getStock()));

            // 确保所有用户都已经接到优惠券后，设置优惠券推送任务完成时间
            CouponTaskDO couponTaskDO = CouponTaskDO.builder()
                    .id(Long.parseLong(message.getCouponTaskId()))
                    .status(CouponTaskStatusEnum.SUCCESS.getStatus())
                    .completionTime(new Date())
                    .build();
            couponTaskMapper.updateById(couponTaskDO);
        }
    }

    /**
     * 执行优惠券扣减和用户获取优惠券的功能
     */
    private void deductStockAndUserGetCoupon(CouponTemplateExecuteEvent message,Long userBatchSize){
        //先执行优惠券库存扣减操作
        Long result = deductStock(message, userBatchSize);
        if(result < 0L){
            throw new ClientException("优惠券数量已不足");
        }
        //再执行用户优惠券操作
        String userBatchKey = String.format(RedisCacheConstant.COUPON_DISTRIBUTE_BATCH_USER_SAVE_KEY, message.getCouponTaskId());
        List<String> userBatchId = stringRedisTemplate.opsForSet().pop(userBatchKey, userBatchSize);
        if (userBatchId != null) {
            UserGetCoupon(userBatchId, message);
        }
    }

    /**
     * 自旋的原因：优惠券数量可能会不够
     */
    private Long deductStock(CouponTemplateExecuteEvent message,Long userBatchSize){
        //同时可以保护数据库 也可以当数据为0时减少一次访问数据库
        if(userBatchSize <= 0){
            log.info("没票了");
            return 0L;
        }
        Long couponTemplateId= Long.valueOf(message.getCouponTemplateId());
        //进行数据库减少
        Long deducted = couponTemplateMapper.deductCouponTemplateStock(userBatchSize, message.getShopNumber(), couponTemplateId);

        if(!SqlHelper.retBool(deducted)){
            //证明此时优惠券模版数量不够我们期望的数值了
             LambdaQueryWrapper<CouponTemplateDO> queryWrapper = Wrappers.lambdaQuery(CouponTemplateDO.class)
                    .eq(CouponTemplateDO::getShopNumber, message.getShopNumber())
                    .eq(CouponTemplateDO::getId, couponTemplateId);
            CouponTemplateDO couponTemplateDO = couponTemplateMapper.selectOne(queryWrapper);
            return couponTemplateDO.getStock() > 0 ?deductStock(message, Long.valueOf(couponTemplateDO.getStock())) : 0L;
        }
        return deducted;
    }

    /**
     * 用户获取优惠券
     */
    private void UserGetCoupon(List<String> userBatchId,CouponTemplateExecuteEvent message){
        Long couponTemplateId = Long.valueOf(message.getCouponTemplateId());

        // 直接赋值大小就可以，避免它的扩容操作
        List<UserCouponDO> userCouponList = new ArrayList<>(userBatchId.size());
        userBatchId.forEach((UserId) -> {
            UserCouponDO couponDO = UserCouponDO.builder()
                    .couponTemplateId(couponTemplateId)
                    .userId(Long.valueOf(UserId))
                    .receiveTime(new Date())
                    .receiveCount(1)
                    .status(UserCouponStatusEnum.UNUSED.getCode())
                    .validStartTime(new Date())
                    .validEndTime(JSON.parseObject(message.getCouponTemplateConsumeRule()).getDate("validityPeriod"))
                    .build();
            userCouponList.add(couponDO);
        });
        //批量保存用户优惠券数据
        batchUserSave(userCouponList,couponTemplateId,message);
    }

    /**
     * 批量保存用户优惠券数据
     */
    private void batchUserSave(List<UserCouponDO> userCouponList,Long couponTemplateId,CouponTemplateExecuteEvent message){
        // 批量插入
        try {
            List<BatchResult> resultList = userCouponMapper.insert(userCouponList, userCouponList.size());
            // 需要进行一个定时任务的来让用户优惠券进行定时失效
            if(!resultList.isEmpty()){
                List<Long> userId = userCouponList.stream().map(UserCouponDO::getUserId).toList();
                Date endDate = JSON.parseObject(message.getCouponTemplateConsumeRule()).getDate("validityPeriod");
                Long deliverTimeStamp = DateUtil.between(new Date(), endDate, DateUnit.MS);
                UserCouponOnTimeCloseEvent onTimeCloseEvent = UserCouponOnTimeCloseEvent.builder()
                        .userId(userId)
                        .couponTemplateId(Long.valueOf(message.getCouponTaskId()))
                        .build();
                userCouponOnTimeCloseProducer.sendMessage(onTimeCloseEvent,deliverTimeStamp);
            }
        }catch (Exception throwable){
            // 可能会出现批量插入失败的原因是因为有些数据数据库已经存在
            Throwable cause = throwable.getCause();
            if (cause instanceof BatchExecutorException) {
                List<Long> userBatchId = userCouponList.stream().map(UserCouponDO::getUserId).toList();
                List<UserCouponDO> existingUserCoupons = getExistingUserCoupons(userBatchId,couponTemplateId);
                for (UserCouponDO couponDO:existingUserCoupons){
                    Long userId = couponDO.getUserId();
                    Iterator<UserCouponDO> iterator = userCouponList.iterator();

                    while (iterator.hasNext()){
                        if(iterator.next().getUserId().equals(userId)){
                            iterator.remove();
                        }
                    }
                }
                // 进行自旋
                if(CollUtil.isNotEmpty(userCouponList)){
                    batchUserSave(userCouponList, couponTemplateId,message);
                }
            }
        }
    }


    /**
     * 获取已经存在数据库中数据
     */
    private List<UserCouponDO> getExistingUserCoupons(List<Long> userBatchId, Long couponTemplateId) {
        LambdaQueryWrapper<UserCouponDO> queryWrapper = Wrappers.lambdaQuery(UserCouponDO.class)
                .eq(UserCouponDO::getCouponTemplateId, couponTemplateId)
                .in(UserCouponDO::getUserId, userBatchId);

        return userCouponMapper.selectList(queryWrapper);
    }
}
